CrewAI 架构详解

整体架构

CrewAI 的架构围绕两个核心概念构建：Crews（团队） 和 Flows（流程）。

CrewAI
├── Crews（团队模式）
│   ├── Agent（智能体）
│   │   ├── Role（角色）
│   │   ├── Goal（目标）
│   │   ├── Backstory（背景）
│   │   └── Tools（工具）
│   ├── Task（任务）
│   │   ├── Description（描述）
│   │   ├── Expected Output（预期输出）
│   │   └── Agent Assignment（分配智能体）
│   └── Process（执行流程）
│       ├── Sequential（顺序）
│       └── Hierarchical（层级）
│
└── Flows（流程模式）
    ├── @start（启动点）
    ├── @listen（监听器）
    ├── @router（路由器）
    └── State（状态管理）

核心概念

1. Crew（团队）

Crew 是 CrewAI 的顶层容器，代表一个协作的智能体团队：

from crewai import Crew, Process

crew = Crew(
    agents=[agent1, agent2],     # 智能体列表
    tasks=[task1, task2],        # 任务列表
    process=Process.sequential,  # 执行流程
    verbose=True                 # 详细输出
)

# 启动执行
result = crew.kickoff(inputs={"topic": "AI Agents"})

Crew 关键属性：

属性	类型	说明
agents	List[Agent]	团队成员
tasks	List[Task]	要执行的任务
process	Process	执行流程类型
verbose	bool	是否输出详细日志
memory	bool	是否启用记忆
share_crew	bool	是否分享遥测数据

2. Agent（智能体）

Agent 是具有特定角色和能力的 AI 实体：

from crewai import Agent

researcher = Agent(
    role="Senior Data Researcher",
    goal="Uncover cutting-edge developments in AI",
    backstory="You're a seasoned researcher...",
    tools=[search_tool, web_tool],
    verbose=True,
    allow_delegation=True,
    memory=True
)

Agent 关键属性：

属性	说明
role	智能体的角色定义
goal	智能体的目标
backstory	智能体的背景故事
tools	可用工具列表
llm	使用的 LLM
allow_delegation	是否允许委派任务
memory	是否启用记忆
verbose	是否详细输出

3. Task（任务）

Task 定义了智能体需要完成的具体工作：

from crewai import Task

research_task = Task(
    description="Research the topic: {topic}",
    expected_output="A detailed report with findings",
    agent=researcher,
    tools=[search_tool],
    output_file="research.md"
)

Task 关键属性：

属性	说明
description	任务描述（支持模板变量）
expected_output	预期输出格式
agent	执行该任务的智能体
tools	任务专用工具
output_file	输出文件路径
context	前置任务的上下文
async_execution	是否异步执行

4. Process（执行流程）

CrewAI 支持两种执行流程：

Sequential（顺序执行）

任务按定义顺序依次执行：

from crewai import Process

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    process=Process.sequential  # 顺序执行
)

Task 1 → Task 2 → Task 3 → Output

Hierarchical（层级执行）

自动分配管理者协调任务：

crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[task1, task2, task3],
    process=Process.hierarchical,  # 层级执行
    manager_llm=ChatOpenAI(model="gpt-4")
)

        Manager
       /   |   \
   Task1 Task2 Task3

项目结构

使用 crewai create crew 创建的标准项目结构：

my_project/
├── .gitignore
├── pyproject.toml
├── README.md
├── .env                      # 环境变量
└── src/
    └── my_project/
        ├── __init__.py
        ├── main.py           # 入口文件
        ├── crew.py           # Crew 定义
        ├── tools/            # 自定义工具
        │   ├── custom_tool.py
        │   └── __init__.py
        └── config/           # YAML 配置
            ├── agents.yaml   # 智能体配置
            └── tasks.yaml    # 任务配置

装饰器模式

CrewAI 使用装饰器简化 Crew 的定义：

from crewai import Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task

@CrewBase
class MyProjectCrew:
    """My Project Crew"""

    @agent
    def researcher(self) -> Agent:
        return Agent(
            config=self.agents_config['researcher'],
            tools=[SerperDevTool()],
            verbose=True
        )

    @agent
    def writer(self) -> Agent:
        return Agent(
            config=self.agents_config['writer'],
            verbose=True
        )

    @task
    def research_task(self) -> Task:
        return Task(config=self.tasks_config['research_task'])

    @task
    def writing_task(self) -> Task:
        return Task(
            config=self.tasks_config['writing_task'],
            output_file='output.md'
        )

    @crew
    def crew(self) -> Crew:
        return Crew(
            agents=self.agents,  # 自动收集 @agent 装饰的方法
            tasks=self.tasks,    # 自动收集 @task 装饰的方法
            process=Process.sequential,
            verbose=True
        )

执行流程

Crew 执行流程

用户输入
    │
    ▼
┌─────────────────┐
│   Crew.kickoff  │  接收输入参数
└─────────────────┘
    │
    ▼
┌─────────────────┐
│   Task 分配     │  根据 Process 分配任务
└─────────────────┘
    │
    ▼
┌─────────────────┐
│   Agent 执行    │  智能体执行分配的任务
│   ├── 思考      │
│   ├── 使用工具  │
│   └── 生成输出  │
└─────────────────┘
    │
    ▼
┌─────────────────┐
│   结果聚合      │  收集所有任务结果
└─────────────────┘
    │
    ▼
最终输出

Agent 决策循环

┌───────────────────────────────────────┐
│           Agent 决策循环               │
├───────────────────────────────────────┤
│                                       │
│  1. 接收任务                          │
│     │                                 │
│     ▼                                 │
│  2. 思考：分析任务需求                 │
│     │                                 │
│     ▼                                 │
│  3. 决策：选择行动                     │
│     ├── 使用工具                      │
│     ├── 委派给其他 Agent              │
│     └── 直接回答                      │
│     │                                 │
│     ▼                                 │
│  4. 执行：执行选择的行动               │
│     │                                 │
│     ▼                                 │
│  5. 观察：评估结果                     │
│     │                                 │
│     ▼                                 │
│  6. 循环或完成                        │
│                                       │
└───────────────────────────────────────┘

记忆系统

CrewAI 支持多种记忆类型：

记忆类型	说明
Short-term	当前执行的临时记忆
Long-term	跨执行的持久记忆
Entity	关于实体的结构化记忆

crew = Crew(
    agents=[...],
    tasks=[...],
    memory=True,  # 启用记忆
    verbose=True
)

与其他框架架构对比

特性	CrewAI	LangGraph	MetaGPT
架构模式	团队 + 流程	状态图	角色 + SOP
任务分配	Agent 绑定	Node 执行	角色订阅
状态管理	Flow State	Graph State	Environment
执行控制	Process	Edge 条件	消息队列
依赖关系	独立	依赖 LangChain	独立

目录结构（源码）

crewai/
├── __init__.py
├── agent/              # 智能体实现
├── agents/             # 智能体构建器
├── crew.py             # Crew 核心实现
├── task.py             # Task 核心实现
├── flow/               # Flow 实现
│   ├── flow.py         # Flow 核心
│   ├── constants.py
│   └── visualization.py
├── tools/              # 工具系统
├── memory/             # 记忆系统
├── mcp/                # MCP 集成
├── llm.py              # LLM 抽象
├── project/            # 项目装饰器
└── utilities/          # 工具函数

---

下一节：17.2 智能体与任务